package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo23partition {
  def main(args: Array[String]): Unit = {


    val conf: SparkConf = new SparkConf()
      .setMaster("local[8]")
      .setAppName("pr")
      .set("spark.default.parallelism", "20") //shuffle 之后默认分区数

    val sc = new SparkContext(conf)
    /**
      * 读取文件分区数决定因素
      * 1、手动指定 minParitions 切片数就必须比这个值要大
      * 2、通过minParitions 可以读取数据的分区数
      * 3、不指定minParitions 分区数据核mr切片规则一致
      *
      */

    val RDD: RDD[String] = sc.textFile("data/words", 1)

    println(s"RDD:${RDD.getNumPartitions}")


    /**
      * 重分区
      * repartition: 改变rdd的分区数，会产生shuffle
      *
      */

    val rdd2: RDD[String] = RDD.repartition(10)


    println(s"rdd2:${rdd2.getNumPartitions}")

    /**
      * coalesce: shuffle 为false 可以用于减少分区
      *
      * 减少分区可以没有shuffle
      * 增加分区必须要要有shuffle
      *
      *
      * coalesce(15, false) : 一般用于合并小文件
      *
      */

    val rdd3: RDD[String] = rdd2.coalesce(15, false)


    println(s"rdd3:${rdd3.getNumPartitions}")


    /**
      * 使用带shuffle类的算可以改变分区
      * 1、如果不写分区数默认等于前一个rdd的分区数
      * 2、写了分区数之后等于指定的分区数
      * 3、如果指定了spark.default.parallelism  shuffle 之后分区数以参数值为准，没有 指定指定的分区数优先级高
      *
      */


    val rdd4: RDD[(String, Iterable[String])] = rdd3
      .groupBy((line: String) => line.split(",")(4), 2)

    println(s"rdd4:${rdd4.getNumPartitions}")


    val rdd5: RDD[(String, Iterable[String])] = rdd3
      .groupBy((line: String) => line.split(",")(4))
    println(s"rdd5:${rdd5.getNumPartitions}")
  }

}
